package cn.edu.cuit.elena.transaction.query;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.LinkedList;

import cn.edu.cuit.elena.common.SackConstants;
import cn.edu.cuit.elena.db.ConnectionPool;
import cn.edu.cuit.elena.db.DataSourceException;
import cn.edu.cuit.elena.db.dao.AccessDataModel;
import cn.edu.cuit.elena.trace.Tracer;
import cn.edu.cuit.elena.transaction.controller.RunTimeExceptionSignal;
import cn.edu.cuit.elena.transaction.system.UserContext;

public class HiveBaseDao
    implements AccessDataModel, ConnectionPool
{
    static final String HIVE_URL = "jdbc:hive://localhost:10000/default";
    static final String HIVE_USERNAME = "";
    static final String HIVE_PASSWORD = "";

    private static LinkedList<Connection> connections;

    private UserContext userContext;

    static
    {

        try
        {
            Class.forName( "org.apache.hadoop.hive.jdbc.HiveDriver" );
            connections = new LinkedList<Connection>();

            for( int i = 0; i < SackConstants.MAX_CONNECTION_NUM; i++ )
            {
                connections.add( DriverManager.getConnection( HIVE_URL, HIVE_USERNAME, HIVE_PASSWORD ) );
            }
        }
        catch( ClassNotFoundException e )
        {
            Tracer.fatalTrace( HiveBaseDao.class, "Hive static block exception : " + e.getMessage(), e );
            throw new RunTimeExceptionSignal( e );
        }
        catch( SQLException e )
        {
            Tracer.fatalTrace( HiveBaseDao.class, "Hive static block exception : " + e.getMessage(), e );
            throw new RunTimeExceptionSignal( e );
        }
    }

    HiveBaseDao(UserContext userContext)
    {
        this.userContext = userContext;
    }

    HiveBaseDao()
    {

    }

    public static HiveBaseDao get( UserContext userContext )
    {

        HiveBaseDao hive = new HiveBaseDao( userContext );
        userContext.setHiveQuerier( hive );
        return hive;
    }

    public static HiveBaseDao get()
    {
        HiveBaseDao hiveBaseDao = new HiveBaseDao();
        return hiveBaseDao;
    }

    @Override
    public void update( String sql )
    {
        // TODO Auto-generated method stub

    }
    
    public void preparesStatement(String hql) throws Exception{
        Connection connection = null;
        try
        {
            Tracer.debugTrace( HiveBaseDao.class, "Prepare HQL:" + hql, null );
            connection = getConnection();
            Statement statement = connection.prepareStatement( hql );
        }
        
        catch( Exception e )
        {
            Tracer.fatalTrace( HiveBaseDao.class, "Query Exception : " + e.getMessage(), e );
            throw e;
        }
        finally
        {
            returnConnection( connection );
        }
    }

    @Override
    public ResultSet query( String hql ) throws Exception
    {
        Connection connection = null;
        try
        {
            Tracer.debugTrace( HiveBaseDao.class, "Query HQL:" + hql, null );
            connection = getConnection();
            Statement statement = connection.createStatement();
            ResultSet resultSet =  statement.executeQuery( hql );
            return resultSet;
        }
        catch( Exception e )
        {
            Tracer.fatalTrace( HiveBaseDao.class, "Query Exception : " + e.getMessage(), e );
            throw e;
        }
        finally
        {
            returnConnection( connection );
        }
    }

    /**
     * Get connection from connection pool, use <code>synchronized</code> block to guarantee working
     * correctly.
     * */
    public Connection getConnection()
    {
        synchronized( connections )
        {
            if( connections.size() == 0 )
            {
                try
                {
                    for( int i = 0; i < SackConstants.MIN_CONNECTION_NUM; i++ )
                    {
                        connections.add( DriverManager.getConnection( HIVE_URL, HIVE_USERNAME, HIVE_PASSWORD ) );
                    }
                }
                catch( SQLException e )
                {
                    throw new DataSourceException( e.getMessage() );
                }
            }

            return connections.pop();
        }
    }

    /**
     * Get connection from connection pool, use <code>synchronized</code> block to guarantee working
     * correctly.
     * */
    public void returnConnection( Connection connection )
    {
        synchronized( connections )
        {
            if( connections.size() == SackConstants.MAX_CONNECTION_NUM )
            {
                try
                {
                    connection.close();
                }
                catch( SQLException e )
                {
                    e.printStackTrace();
                    throw new DataSourceException( e.getMessage() );
                }
                return;
            }
        }

        connections.add( connection );
    }

    //    public ResultSet queryForMax(Interpreter interpreter){
    //        interpreter.
    //    }
    //
    //    public ResultSet queryForMin( Interpreter interpreter )
    //    {
    //    }
    //
    //    public ResultSet queryForCartesian( Interpreter interpreter )
    //    {
    //
    //    }
}
